package reader

import (
	"context"
	"fmt"
	_ "fmt"
	"github.com/segmentio/kafka-go"
	"go.uber.org/zap"
	"log"
	"v8s-ssp-proxy/internal/configuration"
	"v8s-ssp-proxy/internal/types"
	"v8s/logger"
)

// make a new reader that consumes from topic-A

type KafkaReader struct {
	log    *zap.SugaredLogger
	reader *kafka.Reader
}

func NewKafkaReader(properties configuration.KafkaProperties) *KafkaReader {
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:  []string{"localhost:9092", "localhost:9093", "localhost:9094"},
		GroupID:  properties.GroupID,
		Topic:    properties.RequestTopic,
		MaxBytes: 10e6, // 10MB
	})
	return &KafkaReader{
		log:    logger.Logger().Named("mq_reader").Sugar(),
		reader: r,
	}
}

func (r *KafkaReader) KafkaRead(ctx context.Context, msgChan chan types.SSPPartnerMessage) {
	//c := &consumer{r.log, msgChan, make(chan bool)}

	for {
		m, err := r.reader.ReadMessage(context.Background())
		if err != nil {
			break
		}
		fmt.Printf("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))
	}

	if err := r.reader.Close(); err != nil {
		log.Fatal("failed to close reader:", err)
	}
}
